package com.myflink.day05;

import com.myflink.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

import javax.annotation.Nullable;

/**
 * @author Shelly An
 * @create 2020/9/24 8:48
 */
public class ProcessFunction_Keyed {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 4444)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
                    }
                })
                .assignTimestampsAndWatermarks(
                        //改变wm的生成方式，不减1，到了27就触发
                        //为了演示 用打点  一般用周期性的
                        new AssignerWithPunctuatedWatermarks<WaterSensor>() {
                            private Long maxTs = Long.MIN_VALUE;

                            @Nullable
                            @Override
                            public Watermark checkAndGetNextWatermark(WaterSensor lastElement, long extractedTimestamp) {
                                maxTs = Math.max(maxTs, extractedTimestamp);
                                return new Watermark(maxTs);
                            }

                            @Override
                            public long extractTimestamp(WaterSensor element, long previousElementTimestamp) {
                                return element.getTs() * 1000L;
                            }
                        }
                );

        SingleOutputStreamOperator<Long> resultDS = sensorDS
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, Long>() {
                    private Long triggerTs = 0L;

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Long> out) throws Exception {
                        System.out.println(timestamp + "\t定时器触发");
                    }

                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<Long> out) throws Exception {

                        //为了避免重复注册，重复创建对象，注册定时器的时候，判断一下是否注册过了
                        //这里这是简单的写另一个逻辑，即只有第一个数据才注册定时器，后面的都不注册了
                        if (triggerTs == 0) {
                            ctx.timerService().registerEventTimeTimer(
                                    value.getTs() * 1000L + 5000L
                            );
                            //赋值
                            triggerTs = value.getTs() * 1000L + 5000L;
                        }
                    }
                });

        resultDS.print("result");

        env.execute();
    }
}
